package com.rabbit.mq.config;

import com.rabbit.mq.enums.BindingEnum;
import com.rabbit.mq.enums.ExchangeEnum;
import com.rabbit.mq.enums.QueueEnum;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
import org.springframework.amqp.rabbit.batch.BatchingStrategy;
import org.springframework.amqp.rabbit.batch.SimpleBatchingStrategy;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;

@Configuration
public class RabbitConfig {

    /**
     * 配置rabbitmq事务
     *
     * @param connectionFactory
     * @return
     */
    @Bean("rabbitTransactionManager")
    public RabbitTransactionManager rabbitTransactionManager(ConnectionFactory connectionFactory) {
        return new RabbitTransactionManager(connectionFactory);
    }

    @Bean
    public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        // 开启手动 ack
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);

        //设置批量
        factory.setBatchListener(true);
        factory.setConsumerBatchEnabled(true);//设置BatchMessageListener生效
        factory.setBatchSize(10);//设置监听器一次批量处理的消息数量
        return factory;
    }

    @Bean("rabbitTemplate")
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        // false 匹配不到的消息直接被丢弃
        rabbitTemplate.setMandatory(true);
        return rabbitTemplate;
    }

    /**
     * 事务
     * @return
     */
    @Bean("rabbitTransactedTemplate")
    public RabbitTemplate rabbitTransactedTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTransactedTemplate = new RabbitTemplate(connectionFactory);
        // false 匹配不到的消息直接被丢弃
        rabbitTransactedTemplate.setMandatory(true);
        return rabbitTransactedTemplate;
    }

    /**
     * 异步队列
     * @param rabbitTemplate
     * @return
     */
    @Bean
    public AsyncRabbitTemplate asyncRabbitTemplate(RabbitTemplate rabbitTemplate){
        AsyncRabbitTemplate asyncRabbitTemplate = new AsyncRabbitTemplate(rabbitTemplate);
        // 接收超时时间
        asyncRabbitTemplate.setReceiveTimeout(50000);
        return asyncRabbitTemplate;
    }

    @Bean("batchQueueTaskScheduler")
    public TaskScheduler batchQueueTaskScheduler(){
        TaskScheduler taskScheduler=new ThreadPoolTaskScheduler();
        return taskScheduler;
    }

    //批量处理rabbitTemplate
    @Bean("batchQueueRabbitTemplate")
    public BatchingRabbitTemplate batchQueueRabbitTemplate(ConnectionFactory connectionFactory,
                                                           @Qualifier("batchQueueTaskScheduler") TaskScheduler taskScheduler){

        //!!!重点： 所谓批量， 就是spring 将多条message重新组成一条message, 发送到mq, 从mq接受到这条message后，在重新解析成多条message

        //一次批量的数量
        int batchSize=10;
        // 缓存大小限制,单位字节，
        // simpleBatchingStrategy的策略，是判断message数量是否超过batchSize限制或者message的大小是否超过缓存限制，
        // 缓存限制，主要用于限制"组装后的一条消息的大小"
        // 如果主要通过数量来做批量("打包"成一条消息), 缓存设置大点
        // 详细逻辑请看simpleBatchingStrategy#addToBatch()
        int bufferLimit=1024000; //1000 K
        // 指定时间内收集消息，对消息进行组装
        long timeout=5000;

        //注意，该策略只支持一个exchange/routingKey
        //A simple batching strategy that supports only one exchange/routingKey
        BatchingStrategy batchingStrategy=new SimpleBatchingStrategy(batchSize,bufferLimit,timeout);
        return new BatchingRabbitTemplate(connectionFactory,batchingStrategy,taskScheduler);
    }

    @Bean
    public RabbitAdmin rabbitAdmin(RabbitTemplate rabbitTemplate){
        RabbitAdmin rabbitAdmin = new RabbitAdmin(rabbitTemplate);
        //默认就是true
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }

    /**
     * 初始化交换机
     * @param rabbitAdmin
     * @return
     */
    @Bean("createExchange")
    public Object createExchange(RabbitAdmin rabbitAdmin) {
        // 遍历交换机枚举
        ExchangeEnum.toList().forEach(exchangeEnum -> {
            // 根据交换机模式 生成不同的交换机
            switch (exchangeEnum.getType()) {
                case fanout:
                    rabbitAdmin.declareExchange(new FanoutExchange(exchangeEnum.getExchangeName(),
                            exchangeEnum.isDurable(), exchangeEnum.isAutoDelete()));
                    break;
                case topic:
                    rabbitAdmin.declareExchange(new TopicExchange(exchangeEnum.getExchangeName(),
                            exchangeEnum.isDurable(), exchangeEnum.isAutoDelete()));
                    break;
                case direct:
                    rabbitAdmin.declareExchange(new DirectExchange(exchangeEnum.getExchangeName(),
                            exchangeEnum.isDurable(), exchangeEnum.isAutoDelete()));
                    break;
            }
        });
        return null;
    }

    /**
     * 初始化队列
     * @param rabbitAdmin
     * @return
     */
    @Bean("createQueue")
    public Object createQueue(RabbitAdmin rabbitAdmin) {
        // 遍历队列枚举 将队列注册到spring bean工厂 让spring实现队列的管理
        QueueEnum.toList().forEach(queueEnum -> {
            rabbitAdmin.declareQueue(new Queue(queueEnum.getName(),
                    queueEnum.isDurable(), queueEnum.isExclusive(), queueEnum.isAutoDelete(), queueEnum.getArguments()));
        });
        return null;
    }

    /**
     * 绑定
     * @param rabbitAdmin
     * @return
     */
    @Bean("createBinding")
    public Object createBinding(RabbitAdmin rabbitAdmin) {
        // 遍历队列枚举 将队列绑定到指定交换机
        BindingEnum.toList().forEach(bindingEnum -> {
            // 交换机
            ExchangeEnum exchangeEnum = bindingEnum.getExchangeEnum();
            // queue
            QueueEnum queueEnum = bindingEnum.getQueueEnum();
            // 绑定
            rabbitAdmin.declareBinding(new Binding(
                    // queue名称
                    queueEnum.getName(),
                    Binding.DestinationType.QUEUE,
                    // exchange名称
                    exchangeEnum.getExchangeName(),
                    // queue的routingKey
                    queueEnum.getRoutingKey(),
                    // 绑定的参数
                    bindingEnum.getArguments()));
        });
        return null;
    }
}